package com.boot.study.api

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object SideOutputTest {
  def main(args: Array[String]): Unit = {
    // 创建批处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行
    env.setParallelism(1)
    // 状态后端
    //    env.setStateBackend(new RocksDBStateBackend(""))

    // 从外部命令中提取参数
    // windows启动 nc64.exe -L -p 9000
    // --host 127.0.0.1 --port 9000
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = parameterTool.get("host")
    val port: Int = parameterTool.getInt("port")

    // 接受socket文本流
    val inputStream: DataStream[String] = env.socketTextStream(host, port)
    // 转换处理分组
    val dataStream: DataStream[SensorReading] = inputStream.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
    val highTempStream: DataStream[SensorReading] = dataStream.process(new SpiltProcessFuction(30.0))
    highTempStream.print("high")
    highTempStream.getSideOutput(new OutputTag[(String, Long, Double)]("low")).print("low")
    // 启动任务执行
    env.execute("side output test")
  }
}

// 实现自定义的processFuction进行分流
class SpiltProcessFuction(threshold: Double) extends ProcessFunction[SensorReading, SensorReading] {
  override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
    // 如果当前温度值大于阈值输出到主流
    if (value.temperature > threshold) {
      out.collect(value)
    } else {
      // 输出侧输出流
      ctx.output(new OutputTag[(String, Long, Double)]("low"), (value.id, value.timeStamp, value.temperature))
    }
  }
}

